package com.flink.es;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.flink.jdbc.JDBCTemplate;
import http.DtHttpClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.data.RowData;

import java.util.List;

/**
 * @Author hujf
 * @Date 2022/8/29 10:35
 * @Description TODO
 */
public class HttpSourceFunction extends RichSourceFunction<RowData> implements ResultTypeQueryable<RowData> {

    private String url;
    private String body;
    private String type;
    private String result;
    private static final String HTTP_POST = "POST";
    private static final String HTTP_GET = "GET";
    private transient boolean hasNext;
    private DeserializationSchema<RowData> deserializer;
    private String unionType;
    private String unionUrl;
    private String unionData;

    public HttpSourceFunction(String url, String body, String type, DeserializationSchema<RowData> deserializer, String unionData, String unionType, String unionUrl) {
        this.url = url;
        this.body = body;
        this.type = type;
        this.unionData = unionData;
        this.unionType = unionType;
        this.unionUrl = unionUrl;
        this.deserializer = deserializer;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return deserializer.getProducedType();
    }


    public static HttpSourceFunction.Builder builder() {
        return new HttpSourceFunction.Builder();
    }

    @Override
    public void run(SourceContext<RowData> ctx) throws Exception {
        //查询mysql 得到数据
        List<String> company = JDBCTemplate.getData(unionType,unionUrl,unionData);

        String sx = "{\"query\":{\"bool\":{\"must\":{\"match\":{\"company_name.keyword\":\"%s\"}}}}}";



        //用mysql 去查询es的数据
        try {
            for (String s : company) {
                String format = String.format(sx, s);
                JSONObject jsonObject = null;
                if (StringUtils.isNotBlank(type) && HTTP_GET.equals(type.toUpperCase())) {
                    throw new RuntimeException("only post supported");
                }
                jsonObject = DtHttpClient.post(url, format);

                String hits = jsonObject.getString("hits");
                JSONObject hitsObject = JSONObject.parseObject(hits);
                JSONArray hits1 = hitsObject.getJSONArray("hits");
                for (int i = 0; i < hits1.size(); i++) {
                    JSONObject o = (JSONObject) hits1.get(i);
                    JSONObject source = (JSONObject) o.get("_source");
                    ctx.collect(deserializer.deserialize(source.toString().getBytes()));
                }

            }
        } catch (Throwable t) {
            t.printStackTrace(); // print and continue
        }
    }

    @Override
    public void cancel() {

    }

    public static class Builder {
        private String url;
        private String body;
        private String type;
        private DeserializationSchema<RowData> deserializer;

        private String unionType;
        private String unionUrl;
        private String unionData;

        public String getUnionType() {
            return unionType;
        }

        public Builder setUnionType(String unionType) {
            this.unionType = unionType;
            return this;
        }


        public Builder setUnionUrl(String unionUrl) {
            this.unionUrl = unionUrl;
            return this;
        }


        public Builder setUnionData(String unionData) {
            this.unionData = unionData;
            return this;
        }

        public Builder() {

        }

        public Builder setUrl(String url) {
            this.url = url;
            return this;
        }

        public Builder setBody(String body) {
            this.body = body;
            return this;
        }

        public Builder setType(String type) {
            this.type = type;
            return this;
        }

        public Builder setDeserializer(DeserializationSchema<RowData> deserializer) {
            this.deserializer = deserializer;
            return this;
        }

        public HttpSourceFunction build() {
            if (StringUtils.isBlank(url) || StringUtils.isBlank(body) || StringUtils.isBlank(type)) {
                // throw new IllegalArgumentException("params has null");
            }
            return new HttpSourceFunction(this.url, this.body, this.type, this.deserializer,this.unionData,this.unionType,this.unionUrl);
        }

    }
}
